package com.xinjian.jianba.service.impl;

import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.geo.Circle;
import org.springframework.data.geo.Distance;
import org.springframework.data.geo.GeoResults;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisGeoCommands.DistanceUnit;
import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs;
import org.springframework.data.redis.core.GeoOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;

import com.xinjian.jianba.service.RedisService;
import com.xinjian.jianba.utils.JsonUtil;

@Service
@Transactional(rollbackFor = Exception.class)
public class RedisServiceImpl implements RedisService {

	private static final Logger LOGGER = LoggerFactory.getLogger(RedisServiceImpl.class);

	@Autowired
	protected RedisTemplate<String, String> redisTemplate;

	@Override
	public boolean set(final String key, final String value) throws Exception {
		Assert.hasText(key, "Key is not empty.");
		boolean result = redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				connection.set(serializer.serialize(key), serializer.serialize(value));
				return true;
			}
		});
		return result;
	}

	public String get(final String key) throws Exception {
		Assert.hasText(key, "Key is not empty.");
		String result = redisTemplate.execute(new RedisCallback<String>() {
			@Override
			public String doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				byte[] value = connection.get(serializer.serialize(key));
				return serializer.deserialize(value);
			}
		});
		return result;
	}

	public void del(final String key) throws Exception {
		Assert.hasText(key, "Key is not empty.");

		redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection conn) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				return conn.del(serializer.serialize(key));
			}
		});
	}

	@Override
	public boolean expire(final String key, long expire) {
		return redisTemplate.expire(key, expire, TimeUnit.SECONDS);
	}

	/**
	 * 增加地理位置的坐标
	 *
	 * @author liuhaitao
	 * @param lngfloat
	 *            :经度
	 * @param latfloat
	 *            :纬度
	 * @param key
	 *            :缓存key
	 * @return
	 * @return
	 * @exception
	 * @since JDK 1.8
	 */

	@Override
	public Long geoadd(String geoDb, String key, double lngfloat, double latfloat) {
		return redisTemplate.boundGeoOps(geoDb).geoAdd(new Point(lngfloat, latfloat), key);
	}

	/**
	 * 通过给定的坐标和距离(km)获取范围类其它的坐标信息
	 *
	 * @author liuhaitao
	 * @param lngfloat
	 *            :经度
	 * @param latfloat
	 *            :纬度
	 * @param distance
	 *            :查询范围
	 * @param limit
	 *            :查询数量
	 * @param unit
	 *            :查询单位(m/km)
	 * @return
	 * @exception
	 * @since JDK 1.8
	 */
	@Override
	public GeoResults<GeoLocation<String>> radiusGeo(String geoDb, double lngfloat, double latfloat, double distance, Integer limit,
			DistanceUnit unit) {
		try {
			GeoOperations<String, String> geoOps = redisTemplate.opsForGeo();

			// 设置geo查询参数
			GeoRadiusCommandArgs geoRadiusArgs = GeoRadiusCommandArgs.newGeoRadiusArgs();
			geoRadiusArgs = geoRadiusArgs.includeCoordinates().includeDistance();// 查询返回结果包括距离和坐标
			geoRadiusArgs.sortAscending();
			if (limit != null) {
				geoRadiusArgs.limit(limit);// 限制查询数量
			}
			GeoResults<GeoLocation<String>> radiusGeo = geoOps.geoRadius(geoDb, new Circle(new Point(lngfloat, latfloat), new Distance(
					distance, unit)), geoRadiusArgs);
			LOGGER.info("----radiusGeo----------" + JsonUtil.getJsonString(radiusGeo));
			return radiusGeo;
		} catch (Exception e) {
			e.printStackTrace();
			LOGGER.error("通过坐标[" + lngfloat + "," + latfloat + "]获取范围[" + distance + "km的其它坐标失败]" + ", error[" + e + "]");
		}
		return null;
	}

	/**
	 * 移除地理位置信息
	 *
	 * @author liuhaitao
	 * @param
	 * @return
	 * @exception
	 * @since JDK 1.8
	 */
	@Override
	public Long removeGeo(String geoDb, String key) {
		return redisTemplate.opsForGeo().geoRemove(geoDb, key);
	}

	@Override
	public Long count(String geoDb) {
		return redisTemplate.boundZSetOps(geoDb).size();
	}

}